package demo.cyclicbarrier;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;

import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Element;
import net.sf.ehcache.config.CacheConfiguration;
import net.sf.ehcache.config.Configuration;
import net.sf.ehcache.config.TerracottaConfigConfiguration;
import net.sf.ehcache.config.TerracottaConfiguration;

import org.terracotta.express.util.Barrier;
import org.terracotta.express.util.ExpressTerracottaClusteredInstanceFactory;

public class CyclicBarrierDemo {
  private static final Random   RANDOM        = new Random(System.currentTimeMillis());
  private static final String[] words         = new String[] { "this", "is", "awesome!" };
  private static final String   TC_CONFIG_URL = "localhost:9510";

  // u need to start 3 nodes for this
  public static void main(String[] args) throws Exception {
    CyclicBarrierDemo main = new CyclicBarrierDemo();
    main.execute();
  }

  private void execute() throws Exception {
    Barrier barrier = new ExpressTerracottaClusteredInstanceFactory(TC_CONFIG_URL)
        .createClusteredBarrierAsRoot("clusteredBarrier", words.length);

    CacheManager cacheManager = createCacheManagerWithTerracotta(TC_CONFIG_URL, "demoCacheManager");
    Cache resultCache = createAndAddTerracottaCache(cacheManager, "resultCache");

    System.out.println("Waiting for all nodes to arrive (" + words.length + " total nodes) ...");
    // this is a clustered barrier
    int index = barrier.barrier();
    System.out.println("This is node-" + index);
    // do a computation on each node
    compute(resultCache, index);

    barrier.barrier();
    System.out.println("All workers done! Combining result of all workers (might take some time)...");
    // imitating long computation
    doSleep(10);

    List<Value> result = new ArrayList<Value>();
    // all the nodes share the result in the clustered cache
    for (Object keyObj : resultCache.getKeys()) {
      String key = (String) keyObj;
      Value v = (Value) resultCache.get(key).getValue();
      result.add(v);
    }
    Collections.sort(result);
    System.out.println("Final result, after combining results of all workers... ");
    System.out.println("==================================");
    for (Value value : result) {
      System.out.print(value.word + " ");
    }
    System.out.println("\n==================================");
  }

  private Cache createAndAddTerracottaCache(CacheManager cacheManager, String cacheName) {
    CacheConfiguration cacheConfig = new CacheConfiguration(cacheName, 1000).eternal(true);
    // add terracotta clustered to cache config
    cacheConfig.addTerracotta(new TerracottaConfiguration().clustered(true).coherent(true));
    // add cache to cache-manager
    cacheManager.addCache(new Cache(cacheConfig));
    Cache resultCache = cacheManager.getCache(cacheName);
    return resultCache;
  }

  private CacheManager createCacheManagerWithTerracotta(String tcConfigUrl, String cacheManagerName) {
    Configuration cacheManagerConfig = new Configuration();
    cacheManagerConfig.setDefaultCacheConfiguration(new CacheConfiguration("default", 1000));
    cacheManagerConfig.setName(cacheManagerName);
    TerracottaConfigConfiguration tcConfigConfiguration = new TerracottaConfigConfiguration();
    tcConfigConfiguration.setUrl(tcConfigUrl);
    cacheManagerConfig.addTerracottaConfig(tcConfigConfiguration);
    CacheManager cacheManager = new CacheManager(cacheManagerConfig);
    return cacheManager;
  }

  private void compute(Cache resultCache, int index) {
    System.out.println("Worker-" + index + " is working (may take 10 to 20 secs)... ");
    // imitate long computation...
    doSleep(RANDOM.nextInt(11) + 10);
    Value value = new Value(index, words[index]);
    resultCache.put(new Element("key" + index, value));
    System.out.println("==================================");
    System.out.println("Worker-" + index + ": Computed value: \"" + value + "\"");
    System.out.println("==================================");
  }

  private void doSleep(int secs) {
    try {
      Thread.sleep(secs * 1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  private static class Value implements Serializable, Comparable<Value> {

    private final int    index;
    private final String word;

    public Value(int index, String string) {
      this.index = index;
      this.word = string;
    }

    public String toString() {
      return index + ":" + word;
    }

    public int compareTo(Value o) {
      return this.index - o.index;
    }

  }

}
